package Kafka.Producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class CustomProducerCallback {
    // TODO 带回调函数的异步发送

    /**
     * 回调函数会在producer收到ack时调用，为异步调用，该方法有两个参数，分别是
     * 元数据信息（RecordMetadata）和异常信息（Exception），如果Exception为null，说明消息发送成功，如果Exception不为null，说明消息发送失败。
     * 注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试。
     */
    public static void main(String[] args) throws InterruptedException {
        // 1. 创建kafka生产者的配置对象
        Properties properties = new Properties();

        // 2. 给kafka配置对象添加配置信息
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "master:9092");

        // key,value序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // TODO 生产者提高吞吐量生产经验
        // batch.size：批次大小，默认16K
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        // linger.ms：等待时间，默认0 单位ms
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        // RecordAccumulator：缓冲区大小，默认32M：buffer.memory
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        // compression.type：压缩，默认none，可配置值gzip、snappy、lz4和zstd
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");

        /*
        原理解释：
        kafka中的producer会将数据发送至kafka集群供下游consumer消费。此过程涉及到两个线程分别是main线程以及sender线程
        其中：
        main线程中会有Interceptors（拦截器）和Serializer（序列化器）以及Partitioner（分区器），数据经过上述顺序处理过后，
        将发送到RecordAccumulator（内存缓冲区）中，当batch.size（批次大小，默认16K）或linger.ms（等待时间，默认0ms）满足条件时，
        sender线程会从内存缓冲区中拉取数据。

        ※※ 提高Producer的吞吐量的生产经验：
        1、可通过增加batch.size以及linger.ms的大小来增加producer的吞吐量，但此时延迟会一定程度的增加。
        2、可增大内存缓冲区的大小，但此行为会受内存大小限制
        3、可将sender线程发送给kafka集群（broker）的数据进行压缩，但压缩解压缩会消耗一定时间
         */

        // 3. 创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        // 4. 调用send方法,发送消息
        for (int i = 0; i < 5; i++) {
            // 添加回调
            // 该方法在Producer收到ack时调用，为异步调用
            kafkaProducer.send(new ProducerRecord<>("firstTopic", "ansentai " + i),
                    (metadata, exception) -> {
                        if (exception == null) {
                            // 没有异常,输出信息到控制台
                            System.out.println("主题：" + metadata.topic() + "->" + "分区：" + metadata.partition());
                        } else {
                            // 出现异常打印
                            exception.printStackTrace();
                        }
                    });

            // 延迟一会会看到数据发往不同分区
            Thread.sleep(2);
        }

        // 5. 关闭资源
        kafkaProducer.close();
    }
}
